package com.wtwd.campus.rocketmq;

import com.wtwd.campus.config.ConsumerConfig;
import com.wtwd.campus.config.DefaultConsumerConfigure;
import com.wtwd.campus.entity.MqMessage;
import com.wtwd.campus.receiver.IReceiverHandler;
import com.wtwd.campus.receiver.ReceiveMsgServiceContext;
import com.wtwd.campus.utils.ByteObjectUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.List;

@Configuration
public class ReceiveRocketMqMessage extends DefaultConsumerConfigure implements ApplicationListener<ApplicationReadyEvent> {

    Logger log = LoggerFactory.getLogger(ReceiveRocketMqMessage.class);

    @Autowired
    private ConsumerConfig consumerConfig;

    @Autowired
    ReceiveMsgServiceContext receiveMsgServiceContext;

    @Override
    public void onApplicationEvent(ApplicationReadyEvent arg0) {
        try {
            super.listener(consumerConfig.getMangerBackStageTopic(), consumerConfig.getMangerBackStageTag());
        } catch (MQClientException e) {
            log.error("消费者监听器启动失败", e);
        }

    }

    @Override
    public ConsumeConcurrentlyStatus dealBody(List<MessageExt> msgs) {
        for (MessageExt msg : msgs) {
            try {
                MqMessage mqMessage =(MqMessage) ByteObjectUtils.toObject(msg.getBody());
                log.info("接收到队列消：" + mqMessage);
                IReceiverHandler receiverHandler = receiveMsgServiceContext.getIReceiverHandler(mqMessage.getType());
                receiverHandler.handlerMessage(mqMessage);
            } catch (Exception e) {
                if(e instanceof ClassNotFoundException || e instanceof IOException){
                    log.error("MQ消息转对象失败");
                }else{
                    log.error("处理上报逻辑失败：{}",e);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
